package com.shengzai.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo21Cache {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local[4]")
    conf.setAppName("join")

    val sc = new SparkContext(conf)

    /**
     * 统计男生女生的人数
     * 统计班级人数
     * 统计各个年龄的人数
     */
    val scoreRDD: RDD[String] = sc.textFile("data/students.txt")

    val mapRDD: RDD[(String, String, Int, String, String)] = scoreRDD.map(
      line => {
        val split: Array[String] = line.split(",")
        (split(0), split(1), split(2).toInt, split(3), split(4))
      }
    )
    //统计男生女生的人数
   mapRDD.map(
     line =>{
       (line._4,1)
     }
   ).reduceByKey(
     (x,y) => x+y
   ).foreach(println)

    //统计各班级的人数
    mapRDD.map(
      line =>{
        (line._5,1)
      }
    ).reduceByKey(
      (x,y) =>x+y
    ).foreach(println)

    //统计各个年龄的人数
    mapRDD.map(
      line =>{
        (line._3,1)
      }
    ).reduceByKey(
      (x,y) =>x+y
    ).foreach(println)

    /**
     * 上述方法会反复调用mapRDD，mapRDD执行了三次，可以将mapRDD添加至缓存，提升效率
     */
    //将mapRDD添加进缓存
    // mapRDD.cache()是mapRDD.persist(StorageLevel.MEMORY_ONLY)的简写
    mapRDD.persist(StorageLevel.MEMORY_ONLY)
    //统计男生女生的人数
    /**
     * 默认情况下，入股一个RDD被多次使用，会计算多次
     *
     * 缓存级别
     * StorageLevel.MEMORY_ONLY： 仅内存
     * StorageLevel.MEMORY_AND_DISK_SER: 将数据压缩之后方内，内存放不下在方磁盘，
     * 为甚不选择MEMORY_AND_DISK：应该不管压缩不压缩，内存都比磁盘快，所以尽量将数据压缩之后放内存
     *
     * 缓存级别选择和数据量有关，如果内存放不下选择使用MEMORY_AND_DISK_SER
     */
    mapRDD.map(
      line => {
        (line._4, 1)
      }
    ).reduceByKey(
      (x, y) => x + y
    ).foreach(println)

    //统计各班级的人数
    mapRDD.map(
      line => {
        (line._5, 1)
      }
    ).reduceByKey(
      (x, y) => x + y
    ).foreach(println)

    //统计各个年龄的人数
    mapRDD.map(
      line => {
        (line._3, 1)
      }
    ).reduceByKey(
      (x, y) => x + y
    ).foreach(println)
  }
}
